package cn.ac.iscas.service.task.dameng;

import ch.qos.logback.classic.Logger;
import cn.ac.iscas.base.utils.SpringUtil;
import cn.ac.iscas.config.MiddlePlatformConfig;
import cn.ac.iscas.domain.NodeTask;
import cn.ac.iscas.domain.source.DamengSourceColumn;
import cn.ac.iscas.service.component.ComponentService;
import cn.ac.iscas.service.queue.SqlExecuteQueueService;
import cn.ac.iscas.service.queue.dameng.DamengTableQueueService;
import cn.ac.iscas.service.transformer.dameng.*;
import cn.ac.iscas.utils.DamengSqlUtils;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author LJian
 * @version 1.0
 * @description: TODO
 * @date 2022/5/25 9:48
 */
@Service
public class DamengSqlTask extends NodeTask {
    private String nodeType;
    private String kind;
    private String taskId;
    private Map<String, Object> parameters;

    public DamengSqlTask() {

    }

    /**
     * @description:
     * @param: builder
     * @return:
     * @author LJian
     * @date: 2022/12/27 9:09
     */
    private DamengSqlTask(Builder builder) {
        super(builder.jobId, builder.taskId, builder.dependences);
        this.kind = builder.kind;
        this.nodeType = builder.nodeType;
        this.parameters = builder.parameters;
    }

    /**
     * @author LJian
     * @description:
     * @param: null
     * @return:
     * @date: 2022/12/27 9:00
     */
    public static class Builder {
        private String jobId;
        private String taskId;
        private Set<String> dependences;
        private String nodeType;
        private String kind;
        private Map<String, Object> parameters;

        public Builder(String jobId, String taskId, Set<String> dependences) {
            this.jobId = jobId;
            this.taskId = taskId;
            this.dependences = dependences;
        }

        public Builder jobId(String jobId) {
            this.jobId = jobId;
            return this;
        }

        public Builder dependences(Set<String> dependences) {
            this.dependences = dependences;
            return this;
        }

        public Builder nodeType(String nodeType) {
            this.nodeType = nodeType;
            return this;
        }

        public Builder kind(String kind) {
            this.kind = kind;
            return this;
        }

        public Builder parameters(Map<String, Object> parameters) {
            this.parameters = parameters;
            return this;
        }

        public DamengSqlTask build() {
            return new DamengSqlTask(this);
        }

    }


    /**
     * @description:
     * @param: nodeType
     * @param: kind
     * @param: parameters
     * @param: jobId
     * @param: taskId
     * @param: dependences
     * @return:
     * @author LJian
     * @date: 2022/12/27 9:09
     */
    public DamengSqlTask(String nodeType, String kind, Map<String, Object> parameters, String jobId, String taskId, Set<String> dependences) {
        super(jobId, taskId, dependences);

        this.parameters = parameters;
        this.nodeType = nodeType;
        this.kind = kind;

    }

    @Override
    public Map<String, Object> doTask() throws Exception {
        Logger logger = getLogger().getLogger(getJobId());
        logger.info("Do dameng task: {}", getTaskId());
        SqlExecuteQueueService sqlExecuteQueueService = SpringUtil.getBean(SqlExecuteQueueService.class);
        DamengTableQueueService damengTableQueueService = SpringUtil.getBean(DamengTableQueueService.class);
        MiddlePlatformConfig middlePlatformConfig = SpringUtil.getBean(MiddlePlatformConfig.class);
        ComponentService componentService = SpringUtil.getBean(ComponentService.class);

        String frontTableName = "";
        List<DamengSourceColumn> frontColumns = new ArrayList<>();
        if (getDependences() != null && !getDependences().isEmpty()) {
            frontTableName = damengTableQueueService.getTableOrViewName(getJobId(), (String) this.getDependences().toArray()[0]);
            frontColumns = damengTableQueueService.getColumns(getJobId(), (String) this.getDependences().toArray()[0], frontTableName);
        }

        List<String> sourceTables = new ArrayList<>(10);
        List<DamengSourceColumn> sourceColumns = new ArrayList<>();
        if (nodeType.equalsIgnoreCase("reader")) {
            String dataModel = (String) parameters.getOrDefault("tableName", "");
            //parameters.clear();
            parameters.putAll(componentService.getDataSourceByDataModel(dataModel));
            //根据数据模型获取的数据源只包含数据库级，数据表名=模型名称
            //String tableWithSchemaName = parameters.get("schemaName").toString() + "." + parameters.get("tableName").toString();
            String tableWithSchemaName = String.format("\"%s\".\"%s\"", parameters.get("schemaName").toString(), parameters.get("tableName").toString());
            parameters.put("table", tableWithSchemaName);
            Map<String, List<DamengSourceColumn>> tableColumns = new ConcurrentHashMap<>(8);
            List<Map> currentColumns = (List<Map>) parameters.get("tableMeta");
            for (Map column : currentColumns) {

                DamengSourceColumn damengSourceColumn = new DamengSourceColumn();
                damengSourceColumn.setName(column.get("name").toString());
                damengSourceColumn.setType(column.get("columnType").toString());
                damengSourceColumn.setPrimaryKey(Boolean.parseBoolean(column.get("primary").toString()));
                damengSourceColumn.setNullable(Boolean.parseBoolean(column.get("nullable").toString()));
                damengSourceColumn.setCharMaxLen(Integer.parseInt(column.get("charMaxLen").toString()));
                sourceColumns.add(damengSourceColumn);
            }
            tableColumns.put(tableWithSchemaName, sourceColumns);
            damengTableQueueService.put(this.getJobId(), this.getTaskId(), tableColumns);
            sourceTables.add(tableWithSchemaName);
        }

        if (nodeType.equalsIgnoreCase("transformer")) {
            switch (kind.toLowerCase(Locale.ROOT)) {
//                case "invalid_checker":
//                    String invalidCheckSql = new InvalidCheckerTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
//                    sqlExecuteQueueService.add(this.getJobId(), invalidCheckSql);
//                    break;
//                case "add_constant_columns":
//                    String addConstantColumnsSql = new AddConstantColumnsTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
//                    sqlExecuteQueueService.add(this.getJobId(), addConstantColumnsSql);
//                    break;
//                case "filter_condition":
//                    String filterConditionSql = new FilterConditionTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
//                    sqlExecuteQueueService.add(this.getJobId(), filterConditionSql);
//                    break;
//                case "bucket":
//                    String bucketSql = new BucketTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
//                    sqlExecuteQueueService.add(this.getJobId(), bucketSql);
//                    break;
//                case "select_column":
//                    String selectColumn = new SelectColumnsTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
//                    sqlExecuteQueueService.add(this.getJobId(), selectColumn);
//                    break;
                //数据转换算子
                case "data_format_trans":
                    List<String> dataFormatSql = new DamengDataFormatTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
                    sqlExecuteQueueService.add(this.getJobId(), dataFormatSql);
                    break;
                case "fullfil_null":
                case "null_fill_trans":
                    List<String> fullfilNullSql = new DamengFullfilNullTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
                    sqlExecuteQueueService.add(this.getJobId(), fullfilNullSql);
                    break;
                case "deduplication_trans":
                    List<String> deduplicationSql = new DamengDeduplicationTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
                    sqlExecuteQueueService.add(this.getJobId(), deduplicationSql);
                    break;
//                case "null_data_clean_trans":
//                    String nullDataCleanSql = new CleanNullTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
//                    sqlExecuteQueueService.add(this.getJobId(), nullDataCleanSql);
//                    break;
//                case "normalization_trans":
//                    String normalizationSql = new NormalizationTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
//                    sqlExecuteQueueService.add(this.getJobId(), normalizationSql);
//                    break;
                case "boundary_trans":
                    List<String> boundarySql = new DamengBoundaryValueTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
                    sqlExecuteQueueService.add(this.getJobId(), boundarySql);
                    break;
//                case "invalid_value_del_trans":
//                    String invalidValueDelSql = new InvalidValueDeleteTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
//                    sqlExecuteQueueService.add(this.getJobId(), invalidValueDelSql);
//                    break;
//                case "statistics_trans":
//                    String statisticsSql = new StatisticsTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
//                    sqlExecuteQueueService.add(this.getJobId(), statisticsSql);
//                    break;
//                case "togglecase_trans":
//                    String togglecaseSql = new ToggleCaseTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
//                    sqlExecuteQueueService.add(this.getJobId(), togglecaseSql);
//                    break;
                //数据聚合算子
                case "data_aggregate_trans":
                    List<String> dataAggregateSql = new DamengDataAggregateTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
                    sqlExecuteQueueService.add(this.getJobId(), dataAggregateSql);
                    break;
                //数据聚合算子
                case "data_decomposition_trans":
                    List<String> dataDecompositionSql = new DamengDataDecompositionTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
                    sqlExecuteQueueService.add(this.getJobId(), dataDecompositionSql);
                    break;
                default:
                    break;
            }
        } else if (nodeType.equalsIgnoreCase("writer")) {
            String dataModel = (String) parameters.getOrDefault("tableName", "");
            //parameters.clear();
            parameters.putAll(componentService.getDataSourceByDataModel(dataModel));
            //根据数据模型获取的数据源只包含数据库级，数据表名=模型名称
            String tableWithSchemaName = String.format("\"%s\".\"%s\"", parameters.get("schemaName").toString(), parameters.get("tableName").toString());
            parameters.put("table", tableWithSchemaName);
            Map<String, List<DamengSourceColumn>> tableColumns = new ConcurrentHashMap<>(8);
            List<Map> currentColumns = (List<Map>) parameters.get("tableMeta");
            for (Map column : currentColumns) {
                DamengSourceColumn damengSourceColumn = new DamengSourceColumn();
                damengSourceColumn.setName(column.get("name").toString());
                damengSourceColumn.setType(column.get("columnType").toString());
                damengSourceColumn.setPrimaryKey(Boolean.parseBoolean(column.get("primary").toString()));
                damengSourceColumn.setNullable(Boolean.parseBoolean(column.get("nullable").toString()));
                damengSourceColumn.setCharMaxLen(Integer.parseInt(column.get("charMaxLen").toString()));
                sourceColumns.add(damengSourceColumn);
            }
            tableColumns.put(tableWithSchemaName, sourceColumns);
            damengTableQueueService.put(this.getJobId(), this.getTaskId(), tableColumns);
            sourceTables.add(tableWithSchemaName);


            //String finalInsertWriterSql = String.format("INSERT INTO %s (SELECT * FROM %s) ", tableWithSchemaName, frontTableName);
            if (null != frontTableName && CollectionUtils.isNotEmpty(frontColumns)) {
                String finalInsertWriterSql = DamengSqlUtils.getInsertOrderColumnSql(tableWithSchemaName, frontTableName, frontColumns);
                sqlExecuteQueueService.add(this.getJobId(), finalInsertWriterSql);
            }

        }


        System.out.println("++++++++++++++++++++++" + this.getTaskId());

        String msg = "finish MyNodeTask( " + this.getTaskId() + " )";
        Map<String, Object> result = new HashMap<>();
        result.put("msg", msg);
        return result;
    }
}
